package flink.datastream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.junit.Test;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * DataSource
 *
 * @author binarylei
 * @version 2019-11-28
 */
public class SourceDataStreamTest {

    @Test
    public void testSource() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> ds1 = env.addSource(new MyParallelSourceFunction());

        ds1.print().setParallelism(1);
        env.execute();
    }

    public class MyParallelSourceFunction implements ParallelSourceFunction<String> {

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            while (true) {
                ctx.collect(String.valueOf(new Random().nextInt(100000)));
                TimeUnit.SECONDS.sleep(1);
            }
        }

        @Override
        public void cancel() {

        }
    }

}
